package com.example.mq.mqServer.center;

import com.example.mq.mqServer.core.Binding;
import com.example.mq.mqServer.core.MSGQueue;
import com.example.mq.mqServer.core.exchange.Exchange;
import com.example.mq.mqServer.core.message.Message;
import com.example.mq.mqServer.exception.MqException;

import com.sun.xml.internal.bind.v2.TODO;
import lombok.extern.slf4j.Slf4j;


import java.util.LinkedList;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @BelongsProject: 0.mq
 * @BelongsPackage: com.example.mq.mqServer.data_center
 * @Author : 游在海里的鱼
 * @CreateTime : 2023-12-15  15:45
 * @Version : 1.0
 * @ClassName : 内存数据中心
 * @Description : 将数据库中的数据加载到内存,以提高项目的运行速度
 */
@Slf4j
public class MemoryDataCenter {
    //ConcurrentHashMap是线程安全的哈希表
    //交换机哈希表<交换机name，交换机对象>
    private ConcurrentHashMap<String, Exchange> exchangeName_exchange = new ConcurrentHashMap<>();
    //队列哈希表<队列name，队列对象>
    private ConcurrentHashMap<String, MSGQueue> queueName_queue = new ConcurrentHashMap<>();
    //绑定哈希表<交换机name，<队列name，绑定对象>>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> EQname_binding = new ConcurrentHashMap<>();
    //消息哈希表<消息ID，消息对象>
    private ConcurrentHashMap<String, Message> messageId_message = new ConcurrentHashMap<>();
    //队列所属消息哈希表<队列name，LinkedList<此队列下的消息对象>>
    private ConcurrentHashMap<String, LinkedList<Message>> queueName_messageList = new ConcurrentHashMap<>();
    //队列未回应消息哈希表<队列name，<消息ID，消息对象>>
    private ConcurrentHashMap<String, ConcurrentHashMap<String,Message>> queueName_unACKMessage = new ConcurrentHashMap<>();


    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 加载数据到内存，从数据库中
     * @Describe: TODO
     * @modifier: [public] [实例方法]
     **/
    public void init(DiskDataCenter diskDataCenter) {

        exchangeName_exchange.clear();
        for(Exchange exchange:diskDataCenter.selectAllExchange()){
            addExchange(exchange);
        }

        queueName_queue.clear();
        for(MSGQueue queue:diskDataCenter.selectAllQueue()){
            addQueue(queue);
        }

        EQname_binding.clear();
        for(Binding binding:diskDataCenter.selectAllBinding()){
            addBinding(binding);
        }

        messageId_message.clear();
        queueName_messageList.clear();
        queueName_unACKMessage.clear();
        for(MSGQueue queue:queueName_queue.values()){
            for(Message message: diskDataCenter.selectAllMessage(queue.getName())){
                addMessage(message);
                offerMessage(queue,message);
            }
        }
        log.info("[init] [内存数据中心]初始化成功");
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 关于【交换机】的一系列方法
     * @Describe: 新增，删除，获取交换机
     * @modifier: [public] [实例方法]
    **/
    public Exchange addExchange(Exchange exchange){
        Exchange ret = exchangeName_exchange.put(exchange.getName(),exchange);
        if (ret != null) {
            log.debug("[addExchange] [交换机:"+exchange.getName()+"]新增到[exchangeName_exchange]");
        }
        return ret;
    }
    public Exchange removeExchange(String exchangeName){
        Exchange ret = exchangeName_exchange.remove(exchangeName);
        if (ret != null) {
            log.debug("[removeExchange] [交换机:"+exchangeName+"]从[exchangeName_exchange]删除");
        }
        return ret;
    }
    public Exchange getExchange(String exchangeName){
        Exchange ret = exchangeName_exchange.get(exchangeName);
        if (ret != null) {
            log.debug("[getExchange] [交换机:"+exchangeName+"]从[exchangeName_exchange]获取");
        }
        return ret;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 关于【队列】的一系列方法
     * @Describe: 新增，删除，获取队列
     * @modifier: [public] [实例方法]
    **/
    public MSGQueue addQueue(MSGQueue queue){
        MSGQueue ret = queueName_queue.put(queue.getName(),queue);
        if (ret != null) {
            log.debug("[addQueue] [队列:"+queue.getName()+"]新增到[queueName_queue]");
        }
        return ret;
    }
    public MSGQueue removeQueue(String queueName){
        MSGQueue ret = queueName_queue.remove(queueName);
        if (ret != null) {
            log.debug("[removeQueue] [队列:"+queueName+"]从[queueName_queue]删除");
        }
        return ret;
    }
    public MSGQueue getQueue(String queueName){
        MSGQueue ret = queueName_queue.get(queueName);
        if (ret != null) {
            log.debug("[getQueue] [队列:"+queueName+"]从[queueName_queue]获取");
        }
        return ret;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 关于【绑定】的一系列方法
     * @Describe: 新增，删除，获取某交换机和队列的绑定，获取某一交换机的所有绑定
     * @modifier: [public] [实例方法]
    **/
    public Binding addBinding(Binding binding){
        //1.根据exchangeName获取queue-binding，
        //  如果不存在，就new一个新的queue—binding并put进EQname_binding
        ConcurrentHashMap<String,Binding> queue_binding = EQname_binding.computeIfAbsent(binding.getExchangeName()
                ,K->new ConcurrentHashMap<String,Binding>());
        //2.进行对queue_binding加锁
        synchronized (queue_binding){
        //3.判断队列是否绑定过此交换机
            if(queue_binding.get(binding.getQueueName()) != null){
                new MqException("[MemoryDataCenter] [addBinding] [队列:"+binding.getQueueName()
                        +"]已在[交换机:"+binding.getExchangeName()+"]中绑定").printStackTrace();
                return null;
            }
        //4.交换机，队列和绑定构成键值对并返回binding
            Binding ret = queue_binding.put(binding.getQueueName(),binding);
            if (ret != null) {
                log.debug("[addBinding] [队列:"+binding.getQueueName()+
                        "]与[交换机:"+binding.getExchangeName()+"]已绑定到[EQname_binding]");
            }
            return ret;
        }
    }
    public Binding removeBinding(Binding binding) {
        ConcurrentHashMap<String,Binding>queue_binding = EQname_binding.get(binding.getExchangeName());
        if(queue_binding == null){
            new MqException("[MemoryDataCenter] [removeBinding]删除绑定失败，[队列:"+binding.getQueueName()
                    +"] 没有绑定[交换机:"+binding.getExchangeName()+"]").printStackTrace();
        }
        Binding ret = queue_binding.remove(binding.getQueueName());
        if (ret != null) {
            log.debug("[removeBinding] [绑定:"+binding.getQueueName()+
                    "]解绑[交换机:"+binding.getExchangeName()+"]从[EQname_binding]");
        }
        return ret;
    }
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String,Binding>queue_binding = EQname_binding.get(exchangeName);
        if(queue_binding == null){
            return null;
        }
        Binding ret = queue_binding.get(queueName);
        if (ret != null) {
            log.debug("[getBinding] [绑定"+queueName+
                    "]和[交换机"+exchangeName+"]的绑定从[EQname_binding]获取");
        }
        return ret;
    }
    public ConcurrentHashMap<String,Binding> getAllBinding(String exchangeName){
        ConcurrentHashMap<String,Binding> ret = EQname_binding.get(exchangeName);
        if(ret != null){
            log.debug("[getBinding] 获取[交换机"+exchangeName+"]的[所有绑定]从[EQname_binding]");
        }
        return ret;
    }


    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 操作消息
     * @Describe: 增，删，查单个
     * @modifier: [public] [实例方法]
    **/
    public Message addMessage(Message message){
        Message ret = messageId_message.put(message.getId(),message);
        if (ret != null) {
            log.debug("[addMessage] [消息"+message.getId()+"]新增到[messageId_message]");
        }
        return ret;
    }
    public Message removeMessage(String messageId){
        
        Message ret = messageId_message.remove(messageId);
        if (ret != null) {
            log.debug("[removeMessage] [消息"+messageId+"]从[messageId_message]删除");
        }
        return ret;
    }
    public Message getMessage(String messageId){
        
        Message ret = messageId_message.get(messageId);
        if (ret != null) {
            log.debug("[getMessage] [消息"+messageId+"]从[messageId_message]获取");
        }
        return ret;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 操作队列中的消息
     * @Describe: 入，出，查数量
     * @modifier: [public] [实例方法]
    **/
    public boolean offerMessage(MSGQueue queue,Message message){
        //1.根据queueName查询此队列下的消息链表，若不存在，则创建一个新的消息链表
        LinkedList<Message> messages = queueName_messageList.computeIfAbsent(queue.getName()
                ,K->new LinkedList<>());
        //2.新增消息
        synchronized (messages){
            messages.add(message);
        }
        //3.总消息哈希表中也插入此消息，若此消息已存在，这个操作也对总消息哈希表也没有影响
        addMessage(message);
        log.debug("[offerMessage] [消息"+message.getId()+"]offer[队列"+queue.getName()+"]到[queueName_messageList]");
        return true;
    }
    public Message pollMessage(MSGQueue queue)  {
        LinkedList<Message> messages = queueName_messageList.get(queue.getName());
        if(messages == null){
            new MqException("[MemoryDataCenter] [pollMessage]" +
                    "从[队列"+queue.getName() +"]失败,因为队列为空").printStackTrace();
        }
        Message ret = null;
        synchronized (messages){
            if(messages.size() == 0){
                new MqException("[MemoryDataCenter] [pollMessage]" +
                        "从[队列"+queue.getName() +"]失败,因为队列为空").printStackTrace();
            }
            ret = messages.remove(0);
        }
        removeMessage(ret.getId());
        log.debug("[pollMessage] [消息"+ret.getId()+"]poll从[队列"+queue.getName()+"]从[queueName_messageList]");
        return ret;
    }

    public int sizeMessage(MSGQueue queue)  {
        LinkedList<Message> messages = queueName_messageList.get(queue.getName());
        if(messages == null ){
            return 0;
//            throw new MqException("[MemoryDataCenter] [获取消息个数]从[队列"+queue.getName() +"]失败,因为队列为空");
        }
        synchronized (messages){
//            if(messages.size() == 0){
//                throw new MqException("[MemoryDataCenter] [获取消息个数]从[队列"+queue.getName() +"]失败,因为队列为空");
//            }
            log.debug("[sizeMessage] [队列"+queue.getName()+"]有"+messages.size()+"个消息");
            return messages.size();
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 操作未应答消息
     * @Describe: 增，删，查单个
     * @modifier: [public] [实例方法]
    **/
    public Message addUnAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageId_message = queueName_unACKMessage.computeIfAbsent(queueName
                ,K->new ConcurrentHashMap<>());
        Message ret = messageId_message.put(message.getId(),message);
        if (ret != null) {
            log.debug("[addUnAck] [未回应消息"+message.getId()+"]新增到[queueName_unACKMessage]");
        }
        return ret;
    }

    public Message removeUnACK(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageId_message = queueName_unACKMessage.get(queueName);
        if(messageId_message == null){
            return null;
        }
        Message ret = messageId_message.remove(messageId);
        if (ret != null) {
            log.debug("[removeUnACK] [未回应消息"+messageId+"]从[queueName_unACKMessage]删除");
        }
        return ret;
    }

    public Message getUnACK(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageId_message = queueName_unACKMessage.get(queueName);
        if(messageId_message == null){
            return null;
        }
        Message ret = messageId_message.get(messageId);
        if (ret != null) {
            log.debug("[getUnACK] [未回应消息"+messageId+"]从[queueName_unACKMessage]获取");
        }
        return ret;
    }

}
